package com.xuexi.hbase;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

/**
 * 1、 建立用户表和角色表。 用户可以分配多个角色，并且分配给该用户的角色，有优先级。 查询一个角色下所有用户列表 删除用户所分配的其中一个角色。
 * 2、建立部门表。 部分分为父级部门和子级部门。首先可以查询所有顶级部门列表。 查询某个部门下的子部门列表和它的父部门
 * 
 * @author root
 *
 */
public class HbaseTest {

	public static String TN = "phone_cdr";
	public Random r = new Random();
	SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
	private HBaseAdmin admin;
	private HTable htable;

	@Test
	public void createTable() throws Exception {
		if (admin.tableExists(TN)) {
			admin.disableTable(TN);
			admin.deleteTable(TN);
		}
		HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(TN));
		HColumnDescriptor cf = new HColumnDescriptor("cf1");
		cf.setMaxVersions(1);
		cf.setInMemory(true);
		cf.setBlockCacheEnabled(true);
		desc.addFamily(cf);
		admin.createTable(desc);
	}

	/**
	 * 为了查询数据时使用时间的降序排序： rowkey： 186xxxx_Integer.maxvalue-time
	 * 
	 * @throws Exception
	 */
	@Test
	public void insert1() throws Exception {
		String rowkey = "18684640986_20151004161429";
		Put put = new Put(rowkey.getBytes());
		put.add("cf1".getBytes(), "dist_phone".getBytes(), "1390000000".getBytes());
		put.add("cf1".getBytes(), "type".getBytes(), "0".getBytes());
		put.add("cf1".getBytes(), "city".getBytes(), "beijing".getBytes());
		htable.put(put);

	}

	/**
	 * 为了查询数据时使用时间的降序排序： rowkey： 186xxxx_Integer.maxvalue-time
	 * 
	 * @throws Exception
	 */
	@Test
	public void insert2() throws Exception {
		for (int i = 0; i < 10; i++) {
			String phone = getPhone("186");
			List<Put> puts = new ArrayList<Put>();
			for (int j = 0; j < 100; j++) {
				String date = getDatetime();
				long datetime = sdf.parse(date).getTime();
				String rowkey = phone + "_" + (Long.MAX_VALUE - datetime);
				System.out.println(rowkey);
				Put put = new Put(rowkey.getBytes());
				put.add("cf1".getBytes(), "dist_phone".getBytes(), getPhone("139").getBytes());
				put.add("cf1".getBytes(), "type".getBytes(), (r.nextInt(2) + "").getBytes());
				put.add("cf1".getBytes(), "city".getBytes(), "beijing".getBytes());
				put.add("cf1".getBytes(), "time".getBytes(), date.getBytes());
				puts.add(put);
			}
			htable.put(puts);
		}
	}

	/**
	 * 查询2015年11月分18606792191 的通话记录
	 */
	@Test
	public void find() throws Exception {
		Scan scan = new Scan();
		String minrowkey = "18606792191" + "_" + (Long.MAX_VALUE - sdf.parse("20151130235959").getTime());
		scan.setStartRow(minrowkey.getBytes());
		String maxrowkey = "18606792191" + "_" + (Long.MAX_VALUE - sdf.parse("20151101000000").getTime());
		scan.setStopRow(maxrowkey.getBytes());

		ResultScanner rss = htable.getScanner(scan);
		for (Result rs : rss) {
			System.out.println(new String(rs.getColumnLatestCell("cf1".getBytes(), "time".getBytes()).getValue()));
			System.out
					.println(new String(rs.getColumnLatestCell("cf1".getBytes(), "dist_phone".getBytes()).getValue()));
			System.out.println("_---------------------");
		}
	}

	@Test // 查询18606792191 手机所有主叫（1）通话记录
	public void find2() throws Exception {
		Scan scan = new Scan();
		PrefixFilter f1 = new PrefixFilter("18606792191_".getBytes());
		SingleColumnValueFilter f2 = new SingleColumnValueFilter("cf1".getBytes(), "type".getBytes(), CompareOp.EQUAL,
				"1".getBytes());
		FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
		list.addFilter(f1);
		list.addFilter(f2);
		scan.setFilter(list);
		ResultScanner rss = htable.getScanner(scan);
		for (Result rs : rss) {
			System.out.println(new String(rs.getColumnLatestCell("cf1".getBytes(), "time".getBytes()).getValue()));
			System.out
					.println(new String(rs.getColumnLatestCell("cf1".getBytes(), "dist_phone".getBytes()).getValue()));
			System.out.println(new String(rs.getColumnLatestCell("cf1".getBytes(), "type".getBytes()).getValue()));
			System.out.println("_---------------------");
		}
	}

	public String getPhone(String prefix) {
		return prefix + String.format("%08d", r.nextInt(99999999));
	}

	public String getDatetime() {
		return "2015" + String.format("%02d%02d%02d%02d%02d",
				new Object[] { r.nextInt(12) + 1, r.nextInt(30) + 1, r.nextInt(24), r.nextInt(60), r.nextInt(60) });
	}

	@Test
	public void get1() throws Exception {
		String rowkey = "18684640986_20151004161429";
		Get get = new Get(rowkey.getBytes());
		// get.addColumn("cf1".getBytes(), "city".getBytes());
		Result result = htable.get(get);
		// Cell c =result.getColumnLatestCell("cf1".getBytes(),
		// "type".getBytes());
		// System.out.println(new String(c.getValue()));

		result.getColumnCells("cf1".getBytes(), "type".getBytes()).get(0);

	}

	@Before
	public void setup() throws Exception {
		Configuration config = new Configuration();
		config.set("hbase.zookeeper.quorum", "hadoop,hadoop11,hadoop12");
		admin = new HBaseAdmin(config);
		htable = new HTable(config, TN);
	}

	@After
	public void after() throws Exception {
		if (admin != null) {
			admin.close();
		}
		if (htable != null) {
			htable.close();
		}
	}

}
